package com.world.rickcloudendpoint.service.net.process;

import com.world.rickcloudendpoint.common.constant.GlobalConstants;
import com.world.rickcloudendpoint.common.ThreadNameFactory;
import com.world.rickcloudendpoint.message.AbstractNetMessage;
import com.world.rickcloudendpoint.common.utils.ExecutorUtil;
import lombok.extern.log4j.Log4j2;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.*;

/**
 * @description: 队列消息执行处理器
 * @author: jlsong
 * @date: 2021/12/23 21:43:07
 * @version: 1.0
 */
@Log4j2
public class QueueMessageExecutorProcessor implements IMessageProcessor {

    /**
     * 消息队列
     */
    protected final BlockingQueue<AbstractNetMessage> queue;

    /**
     * 消息处理线程停止时剩余的还未处理的消息
     */
    private volatile List<AbstractNetMessage> leftQueue;

    /**
     * 消息处理线程池
     */
    private volatile ExecutorService executorService;

    /**
     * 线程池的县城个数
     */
    private final int executorCoreSize;

    /**
     * 是否停止
     */
    private volatile boolean stop = false;

    /**
     * 处理消息总数
     */
    public long statisticsMessageCount = 0;

    /**
     * 是否决定需要将未处理的消息放入剩余队列中
     */
    private final boolean processLeft;

    public QueueMessageExecutorProcessor() {
        this(false, 1);
    }

    public QueueMessageExecutorProcessor(boolean processLeft, int executorCoreSize) {
        queue = new LinkedBlockingDeque<>();
        this.processLeft = processLeft;
        this.executorCoreSize = executorCoreSize;
    }

    @Override
    public synchronized void start() {
        if (this.executorService != null) {
            throw new IllegalStateException("The executorService has not been stopped.");
        }
        stop = false;
        ThreadNameFactory factory = new ThreadNameFactory(GlobalConstants.Thread.GAME_MESSAGE_QUEUE_EXCUTE);
        this.executorService = Executors.newFixedThreadPool(this.executorCoreSize, factory);

        for (int i = 0; i < this.executorCoreSize; i++) {
            this.executorService.execute(new Worker());
        }

        log.info("Message processor executorService started [ {} with {} threads ]", this.executorService, this.executorCoreSize);
    }

    @Override
    public void stop() {
            log.info("Message processor executor {} stopping...", this);
            this.stop = true;
            if (this.executorService != null){
                ExecutorUtil.shutdownAndAwaitTermination(this.executorService, 50, TimeUnit.MILLISECONDS);
                this.executorService = null;
            }
            log.info("Message processor executor {}  stopped.", this);
            if (this.processLeft){
                // 将未处理的消息放入到leftQueue中，以备后续处理
                this.leftQueue = new LinkedList<AbstractNetMessage>();
                while (true){
                    AbstractNetMessage _msg = this.queue.poll();
                    if (_msg != null){
                        this.leftQueue.add(_msg);
                    }else {
                        break;
                    }
                }
            }
            this.queue.clear();
    }

    @Override
    public void put(AbstractNetMessage msg) {
        try {
            queue.put(msg);
            log.debug("put queue size: {} ", queue.size());
        } catch (InterruptedException e) {
            log.error("put message into queue failed,,the reason is {} ", e.getMessage(), e);
        }
    }

    /**
     * 达到5000上限时认为满了
     */
    @Override
    public boolean isFull() {
        return this.queue.size() > 5000;
    }

    /**
     * 处理具体的消息，每个消息有自己的参数和来源,如果在处理消息的过程中发生异常,则马上将此消息的发送者断掉
     *
     * @param msg
     */
    public void process(AbstractNetMessage msg) {
        if (msg == null) {
            log.warn("The message that got  from queue is null.");
            return;
        }

        long begin = System.nanoTime();
        this.statisticsMessageCount++;
        try {
            //TODO
        } catch (Exception e) {
            log.error("process message of queue failed,the reason is  {} ", e.getMessage(), e);
        } finally {
            long time = (System.nanoTime() - begin) / (1000 * 1000);
            if (time > 1) {
                log.info("dispatch message id: {}  time: {}  ms total: {}", msg.getNetMessageHead().getCmd(), time, this.statisticsMessageCount);
                ;
            }
        }
    }

    /**
     * 取得消息处理器停止后的遗留的消息
     *
     * @return 遗留的消息
     */
    public List<AbstractNetMessage> getLeftQueue() {
        return leftQueue;
    }

    /**
     * 重置
     */
    public void resetLeftQueue() {
        this.leftQueue = null;
    }

    private final class Worker implements Runnable {

        @Override
        public void run() {
            while (!stop) {
                try {
                    process(queue.take());
                    log.debug("run queue size: {} ", queue.size());
                } catch (InterruptedException e) {
                    log.warn("[#CORE.QueueMessageExecutorProcessor.run] [Stop process]");
                    break;
                } catch (Exception e) {
                    log.error("process message from queue failed,the reason is {} ", e.getMessage(), e);
                }
            }
        }
    }
}
